package HBaseIA.TwitBase.myTest;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.joda.time.DateTime;
import utils.Md5Utils;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;

/**
 * RowKey设计
 */
public class RowKeyDesign {

    public static final byte[] TABLE_NAME = Bytes.toBytes("twits");
    public static final byte[] TWITS_FAM  = Bytes.toBytes("twits");
    public static final byte[] USER_COL   = Bytes.toBytes("user");
    public static final byte[] TWIT_COL   = Bytes.toBytes("twit");
    private static final int longLength = 8; // bytes
    private static final TableName twits = TableName.valueOf("twits");


    public static void main(String[] args) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "hadoop103,hadoop104,hadoop105");

        //建表
        Connection conn = ConnectionFactory.createConnection(conf);
        Admin admin = conn.getAdmin();
        ColumnFamilyDescriptor cfd = ColumnFamilyDescriptorBuilder
                .newBuilder("twits".getBytes(StandardCharsets.UTF_8))//列族名
                .build();
        TableDescriptor td = TableDescriptorBuilder
                .newBuilder(twits)//表名
                .setColumnFamilies(Collections.singleton(cfd))
                .build();
        admin.createTable(td);
        admin.close();

        //rowKey设计:用户唯一标识+时间戳
        byte[] userHash = Md5Utils.md5sum("abc");//length为16,将用户唯一标识散列化
        byte[] timestamp = Bytes.toBytes(-1 * 1631784225309L);//length为8
        //设置rowKey长度16+8 MD5字符串长度+时间戳长度
        byte[] rowKey = new byte[Md5Utils.MD5_LENGTH + longLength];
        //组装rowKey
        int offset = 0;
        //将数组userHash放入rowKey数组的高位
        offset = Bytes.putBytes(rowKey, offset, userHash, 0, userHash.length);
        System.out.println(offset);//16
        //将timestamp数组放入rowKey数组中userHash的后面
        Bytes.putBytes(rowKey, offset, timestamp, 0, timestamp.length);

        //存值
        Put put = new Put(rowKey);
        put.add(TWITS_FAM, USER_COL, Bytes.toBytes("LiLi"));
        put.add(TWITS_FAM, TWIT_COL, Bytes.toBytes("hello word"));
        Table twitsTable = conn.getTable(twits);
        twitsTable.put(put);

        //扫描表
        byte[] userHash1 = Md5Utils.md5sum("abc");
        byte[] startRow = Bytes.padTail(userHash1, longLength); // 212d...866f00...
        byte[] stopRow = Bytes.padTail(userHash1, longLength);
        stopRow[Md5Utils.MD5_LENGTH-1]++;                      // 212d...867000...
        Scan s = new Scan(startRow, stopRow);
        s.addColumn(Bytes.toBytes("twits"), Bytes.toBytes("user"));
        s.addColumn(Bytes.toBytes("twits"), Bytes.toBytes("twit"));
        HTablePool pool = new HTablePool(conf, Integer.MAX_VALUE);
        HTableInterface twits = pool.getTable("twits");
        ResultScanner results = twits.getScanner(s);
        ArrayList<Twit> twitsList = new ArrayList<>();//结果集
        for(Result r : results) {
            Twit twit = new Twit(
                Bytes.toString(r.getColumnLatest(TWITS_FAM, USER_COL).getValue()),
                new DateTime(-1 * Bytes.toLong(Arrays.copyOfRange(r.getRow(), Md5Utils.MD5_LENGTH, Md5Utils.MD5_LENGTH + longLength))),
                Bytes.toString(r.getColumnLatest(TWITS_FAM, TWIT_COL).getValue())
            );
            twitsList.add(twit);
        }
        twits.close();


    }
}
